package com.bw.ods;

import com.alibaba.fastjson.JSONObject;
import com.bw.util.CDCUtil;
import com.bw.util.KafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @BelongsProject: work_new_order
 * @BelongsPackage: com.bw.ods
 * @Author: wuxiaopeng
 * @CreateTime: 2025-07-08  21:51
 * @Description: TODO
 * @Version: 1.0
 */
public class mysql_to_ods {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        DataStreamSource<String> mysql = CDCUtil.getMysqlSource(env, "work_order");

//        mysql.print();

        SingleOutputStreamOperator<String> data = mysql.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                boolean b = JSONObject.isValid(s);
                if (!b)
                    return false;
                if (JSONObject.parseObject(s).getJSONObject("after") == null)
                    return false;
                return true;
            }
        });

        data.print();

        data.addSink(KafkaUtil.getKafkaSink("new_work_order_ods_all_data"));

        env.execute("work_order");
    }
}
